package juc.work;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Executor;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 手写线程池 简单实现
 */
public class MyThreadPool implements Executor {
    // 线程计数
    private final AtomicInteger ctl = new AtomicInteger(0);

    // 线程池参数
    private volatile int corePoolSize = 0;
    private volatile int maximumPoolSize = 0;

    // 阻塞队列
    private final BlockingQueue<Runnable> workQueue;

    public MyThreadPool(int corePoolSize, int maximumPoolSize, BlockingQueue<Runnable> workQueue) {
        this.corePoolSize = corePoolSize;
        this.maximumPoolSize = maximumPoolSize;
        this.workQueue = workQueue;
    }

    // worker线程实现Runnable接口
    private final class Worker implements Runnable {
        // Worker线程需要一个线程Thread对象
        final Thread thread;
        Runnable firstTask;
        // 初始化的第一个任务
        public Worker(Runnable firstTask) {
            // 新线程参数就是Worker对象实例
            this.thread = new Thread(this);
            this.firstTask = firstTask;
        }

        // 从阻塞队列获取任务
        private Runnable getTask() {
            // 一直循环获取
            while(true) {
                try {
                    System.out.println("工作队列长度:" + workQueue.size());
                    return workQueue.take();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }

        @Override
        public void run() {
            Runnable task = firstTask;
            try {
                while(task != null || (task = getTask()) != null) {
                    task.run();
                    // 线程大于最大设置
                    if(ctl.get() > maximumPoolSize)
                        break;
                    task = null;
                }
            } finally {
                ctl.decrementAndGet();
            }
        }
    }

    // 添加工作线程
    private boolean addWorker(Runnable firstTask) {
        if(ctl.get() > maximumPoolSize) return false;
        Worker worker = new Worker(firstTask);
        worker.thread.start();
        ctl.incrementAndGet();
        return true;
    }

    // 拒绝策略
    private void reject() {
        throw new RuntimeException("错误！线程池线程不足");
    }

    // 执行
    @Override
    public void execute(Runnable command) {
        int cnt = ctl.get();
        // 小于核心线程数量直接添加工作线程
        if(cnt < corePoolSize) {
            if(!addWorker(command)){
                reject();
            }
            return;
        }
        // 大于核心线程数就直接添加到阻塞队列，并尝试添加worker线程。队列满了无法添加任务就执行下一步
        if(!workQueue.offer(command)) {
            // 并且线程也满了
            if(!addWorker(command)) {
                reject();
            }
        }
    }

    public static void main(String[] args) {
        MyThreadPool myThreadPool = new MyThreadPool(2, 2, new ArrayBlockingQueue<>(10));
        for (int i = 0; i < 10; i++) {
            int k = i;
            myThreadPool.execute(() -> {
                try {
                    Thread.sleep(100);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println("任务编号 " + k);
            });
        }
    }
}
